package com.easy.consumer.compoment;

import com.alibaba.fastjson.JSON;
import com.easy.common.bean.ConresultBean;
import com.easy.common.bean.OrderBean;
import com.easy.common.bean.SeatBean;
import com.easy.common.exception.ServiceException;
import com.easy.common.service.IConresultService;
import com.easy.common.service.IOrderService;
import com.easy.common.service.ISeatService;
import com.easy.common.util.SKRequest;
import com.easy.common.util.SYS;
import org.apache.dubbo.config.annotation.DubboReference;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.SetOperations;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.concurrent.TimeUnit;

@Component//消费端
public class SKConsumer {

    @Autowired
    RedisTemplate redisTemplate;
    @Autowired//转发用的rabbitTemplate
    RabbitTemplate rabbitTemplate;
    @DubboReference
    IConresultService concertService;
    @DubboReference
    IOrderService orderService;
    @DubboReference
    ISeatService seatService;

    /**
     * 消费者消息处理方法
     * 监听指定队列的消息并进行处理
     *
     * @param message 从队列中接收到的消息内容
     */
    //监听器监听队列，标注监听哪个队列
    //监听者设计模式
    @RabbitListener(queues = SYS.SK_QUEUE)
    public void consumerMessage(String message){
        try {
            //将队列中的JSON数据转换成SKRequest对象
            SKRequest skRequest = JSON.parseObject(message, SKRequest.class);
            //判断是否有座位
            ValueOperations vos = redisTemplate.opsForValue();//处理字符串值
            //获取hash操作对象
            HashOperations hos = redisTemplate.opsForHash();//处理hash值
            SetOperations sos = redisTemplate.opsForSet();
            Integer seatid = skRequest.getConcertid();
            String sk_seat_key = SYS.SK_PRE_SEAT_KEY + seatid;//redis中存储的商品的key值对应val是数量
            //秒杀活动的key
            Integer concertid = skRequest.getConcertid();
            String sk_concert_key = SYS.SK_PRE_CONCERT_KEY + concertid;
            //判断活动是否开启
            boolean hashConcert = redisTemplate.hasKey(sk_concert_key);
            //获取活动状态
            int concert_statues = (int) hos.get(sk_concert_key, SYS.SK_CONCERT_STATUS_HASH_KEY);
            //时间戳
            long concert_start_time = (long) hos.get(sk_concert_key, SYS.SK_CONCERT_STATUS_TIME_KEY);
            long concert_end_time = (long) hos.get(sk_concert_key, SYS.SK_CONCERT_END_TIME_KEY);
            if(!hashConcert||concert_statues!=1 || System.currentTimeMillis() < concert_start_time || System.currentTimeMillis() > concert_end_time){
                //活动不存在, || 活动状态不对  || 活动未开始 ||活动已结束
                // 也是秒杀失败的信息，将数据保存到秒杀记录表中，封装成秒杀记录的数据  ，转发到死信队列中
                rabbitTemplate.convertAndSend(SYS.SK_EXCHANGE,SYS.SK_DLX_ROUTING_KEY,message);
                //结束流程
                return;
            }
            //获取分布式锁
            //某一个秒杀商品的分布式锁的KEY
            String sk_lock_key = SYS.SK_PRE_LOCK_KEY + seatid;
            while (vos.setIfAbsent(sk_lock_key, "",SYS.SK_LOCK_EXPIRE_TIME, TimeUnit.MILLISECONDS)) ;//一直尝试设置，直到设置成功为止
            long a=System.currentTimeMillis();
            System.out.println("获取锁成功"+a);


            //检查用户是否已经抢购   --- 用户允许抢购的数量是可以在抢购商品中设置的
            if(sos.isMember(SYS.SK_RESULT_CONCERT_SEAT_KEY+concertid, skRequest.getUserid())){
                //已经抢购过了
                rabbitTemplate.convertAndSend(SYS.SK_EXCHANGE,SYS.SK_DLX_ROUTING_KEY,message);
                //开放分布式锁
                redisTemplate.delete(sk_lock_key);
                //结束流程
                return;
            }
            //秒杀成功在redis中记录秒杀用户的信息，防止一个人买两个tickets
            sos.add(sk_concert_key, skRequest.getUserid());
            //检查库存，商品数量是否还有,如果数量不足，转发死信，放弃分布式锁，结束流程
            if (vos.get(sk_seat_key) == null || Integer.parseInt(vos.get(sk_seat_key).toString()) <= 0) {
                //数量不够，失败，转发死信队列
                rabbitTemplate.convertAndSend(SYS.SK_EXCHANGE, SYS.SK_DLX_ROUTING_KEY, message);
                redisTemplate.delete(sk_lock_key);
                return;
            }
            //扣数量
            redisTemplate.opsForValue().decrement(sk_seat_key);
            //扣完数量就可以解锁(放开锁)
            long b=System.currentTimeMillis();
            System.out.println("秒杀成功，耗时："+(b-a));
            redisTemplate.delete(sk_lock_key);

            //生成订单--操作数据库--Dubbo服务
            //生成秒杀记录和秒杀订单存储到数据库当中去
            //封装好ConresultBean
            ConresultBean skResult = new ConresultBean();
            skResult.setConcertid(skRequest.getConcertid());
            skResult.setUserid(skRequest.getUserid());
            //skResult.setSeatid(skRequest.getSeatid());
            skResult.setResult(1);
            //skResult.setId(skRequest.getSkId());
            skResult.setDatatime(LocalDateTime.now());
//            concertService.save(skResult);
            //封装订单对象
            OrderBean order = new OrderBean();
            order.setUserid(order.getUserid());
            order.setSeatid(order.getSeatid());
            order.setOrderno(order.getOrderno());
            order.setStatus(order.getStatus());
            order.setCreatetime(order.getCreatetime());
            order.setPaytime(order.getPaytime());
            order.setExpiretime(order.getExpiretime());
            order.setConcertid(order.getConcertid());
            order.setSeatpriceid(order.getSeatpriceid());
            //获取座位数据
//            SeatBean seat = seatService.getOne(order.getSeatid());
//            SeatBean seat=seatService.get
//            //对数据库中的数据-1getSeatById(1);
//            seat.setCount(seat.getCount()-1);
//            seatService.updateSeat(seat);

            return;



        } catch (Exception e) {
        //失败，将消息发送给死信队列，死信队列处理
        rabbitTemplate.convertAndSend(SYS.SK_EXCHANGE,SYS.SK_DLX_ROUTING_KEY,message);
        throw new ServiceException(e.getMessage());
        }
    }


    /**
     * 消费死信队列消息的监听器方法   秒杀失败的消息队列
     * 该方法监听系统死信队列，处理无法正常消费的消息
     *
     * @param message 从死信队列中接收到的消息内容
     */
    @RabbitListener(queues = SYS.SK_DLX_QUEUE)
    public void consumerDLXMessage(String message){
        //在秒杀记录表中添加一条秒杀失败的记录
//
//        SeckillRecord record = new SeckillRecord();
//        record.setUserId(userId);
//        record.setProductId(productId);
//        record.setStatus(SeckillRecord.STATUS_FAILED);
//        record.setCreateTime(new Date());
//        try {
//            seckillRecordMapper.insert(record);
//        } catch (Exception e) {
//            log.error("插入秒杀失败记录异常，userId:{}, productId:{}", userId, productId, e);
//            // 根据业务需求决定是否需要抛出异常或进行其他处理
//        }

    }

}




















